package com.example.bigdata.kafka.producer;


import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 *  自定义拦截器  拦截指定消息、修改消息内容、统计发送量、成功率 等 都需要使用拦截器实现
 *
 */

public class ProducerInterceptorPrefix implements ProducerInterceptor <String,String> {
    private volatile  long sendSuccess = 0;
    private volatile  long sendFailure = 0;

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        String modifiedValue = "prefix1-"+record.value();
        return  new ProducerRecord<>(record.topic(),
                record.partition(),
                record.timestamp(),
                record.key(),
                modifiedValue,record.headers());
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            sendSuccess++;
        } else {
            sendFailure++;
        }
    }

    @Override
    public void close() {
        double successRatio = sendSuccess / (sendFailure + sendSuccess);
        System.out.println("[INFO] 发送成功率 = "
            +String.format("%f",successRatio*100) + "%");

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
